package com.heima.article.lisener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.UpdateArticleMessage;
import com.heima.article.entity.ArticleStreamMessage;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.StringUtils;
import springfox.documentation.spring.web.json.Json;

import java.time.Duration;
import java.util.Date;

@EnableBinding(value = ArticleProcess.class)
public class ArticleListener {
    @Value("${commit.time}")
    private String commitTime;

    @StreamListener(value = "article_behavior")
    @SendTo(value = "article_result")
    public KStream<String, String> process( KStream<String, String> input){
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                System.out.println("传入数据"+value);
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);

                return new KeyValue<>(updateArticleMessage.getArticleId().toString(), value);
            }
        });
        KGroupedStream<String, String> groupedStream = map.groupByKey();
        TimeWindowedKStream<String, String> windowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofMillis(Long.parseLong(commitTime))));
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                return null;
            }
        };
        Aggregator<String, String, String> aggregate = new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                System.out.println("上一次传入的数据："+aggregate);
                ArticleStreamMessage message = null;
                if (StringUtils.isEmpty(aggregate)){
                    message = new ArticleStreamMessage();
                    message.setArticleId(Long.parseLong(key));
                }else {
                    message =  JSON.parseObject(aggregate,ArticleStreamMessage.class);
                }

                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value,UpdateArticleMessage.class);

                switch (updateArticleMessage.getType()){
                    case 0:
                        message.setView(message.getView()+updateArticleMessage.getAdd());
                        System.out.println("阅读量自增"+message);
                        break;
                    case 1:
                        message.setLike(message.getLike()+updateArticleMessage.getAdd());
                        System.out.println("点赞量自增"+message);
                        break;
                    case 2:
                        message.setComment(message.getComment()+updateArticleMessage.getAdd());
                        System.out.println("评论量自增"+message);
                        break;
                    case 3:
                        message.setCollect(message.getCollect()+updateArticleMessage.getAdd());
                        System.out.println("收藏量自增"+message);
                        break;
                }
                String json = JSON.toJSONString(message);
                return json;
            }
        };
        KTable<Windowed<String>, String> aggregate1 = windowedKStream.aggregate(init, aggregate);
        KStream<String, String> map1 = aggregate1.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                System.out.println("聚合信息"+value);
                return new KeyValue<>(key.key(), value);
            }
        });
        return map1;
    }
}
